package b.gate.utils.trying

import java.nio.ByteOrder

import akka.actor.ActorSystem
import akka.event.slf4j.SLF4JLogging
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.{BidiFlow, Flow, Sink, Source}
import akka.util.ByteString

import scala.language.postfixOps

/**
  * 贲国称 于 2016/11/20 创建。
  */
trait Message
case class Ping(id: Int) extends Message
case class Pong(id: Int) extends Message
object TryGraph extends App with SLF4JLogging {
  implicit val system = ActorSystem("biL2R")
  implicit val materializer = ActorMaterializer()
  def toBytes(msg: Message): ByteString = {
    implicit val order = ByteOrder.LITTLE_ENDIAN
    log.info(s"toBytes: $msg")
    msg match {
      case Ping(id) => ByteString.newBuilder.putByte(1).putInt(id).result()
      case Pong(id) => ByteString.newBuilder.putByte(2).putInt(id).result()
    }
  }
  def fromBytes(bytes: ByteString): Message = {
    implicit val order = ByteOrder.LITTLE_ENDIAN
    log.info(s"fromBytes: $bytes")
    val it = bytes.iterator
    it.getByte match {
      case 1 => Ping(it.getInt)
      case 2 => Pong(it.getInt)
      case other => throw new RuntimeException(s"parse error: expected 1|2 got $other")
    }
  }
//  val codec = BidiFlow.fromGraph(GraphDSL.create() { b =>
//    // construct and add the top flow, going outbound
//    val outbound = b.add(Flow[Message].map(toBytes))
//    // construct and add the bottom flow, going inbound
//    val inbound = b.add(Flow[ByteString].map(fromBytes))
//    // fuse them together into a BidiShape
//    BidiShape.fromFlows[Message, ByteString, ByteString, Message](outbound, inbound)
//  })
  val codec = BidiFlow.fromFunctions(toBytes, fromBytes)
  // test it by plugging it into its own inverse and closing the right end
//  val pingpong = Flow[Message].collect { case Ping(id) => log.info(s"ping->pong: $id"); Pong(id+100) }
//  val flow = codec.atop(codec.reversed).join(pingpong)
//  Source((0 to 1).map(Ping)).via(flow).map {
//    msg => log.info(s"pong: $msg")
//  }.runWith(Sink.seq)
  val around = Flow[ByteString].collect { case b: ByteString => log.info(s"ping->pong: $b"); ByteString(b.head.toInt+100) }
  val flow = codec.join(around)
  Source((0 to 1).map(Ping)).via(flow).map {
    msg => log.info(s"pong: $msg")
  }.runWith(Sink.seq)
  Source((10 to 11).map(Pong)).via(flow).map {
    msg => log.info(s"x-ping: $msg")
  }.runWith(Sink.seq)
}
